package cn.tedu.emqx.consumer;

import org.eclipse.paho.client.mqttv3.*;

import java.nio.charset.StandardCharsets;

/**
 * 作为一个监听主题 消息回调的执行者，有一些方法
 * 需要约束（实现接口）
 */
public class EmqxDemoConsumer implements MqttCallbackExtended {
    private MqttClient mqttClient;

    public EmqxDemoConsumer(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }
    /**
     * 异步监听线程，通知消费者，连接已建立完成
     * @param b 是否为ssl加密的连接
     *          true：ssl加密的连接，false：非ssl加密的连接，默认为false
     * @param host 连接的host
     */
    @Override
    public void connectComplete(boolean b, String host) {
        System.out.println("连接对象mqttClient建立连接，连接的host是" + host);
        // 消费者应该在连接建立完成之后订阅自己业务的主题，不同消费者订阅的业务主题应该不同
        // 订阅监听主题，要通过mqttClient对象调用subscribe方法实现
        try {
            mqttClient.subscribe("/test/topic/01");
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 异步监听线程，告知消费者，连接已断开
     * @param throwable 断开的原因
     */
    @Override
    public void connectionLost(Throwable throwable) {
        System.out.println("连接对象mqttClient已断开，原因：" + throwable.getMessage());
    }

    /**
     * 异步监听线程，告知消费者，有一个正在被其监听的主题中产生了一条消息
     * 消费逻辑的核心，通知消费者来源、来源主题（实际情况下一个消费者不会只监听一个主题）
     * @param topic 消息来源主题
     * @param mqttMessage 消息内容
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        System.out.println("消费者收到消息，topic是" + topic);
        byte[] payload = mqttMessage.getPayload();
        String msg = new String(payload, StandardCharsets.UTF_8);
        System.out.println("消息内容：" + msg);

    }

    /**
     * 异步监听线程，告知消费者，某一条消息已经彻底消费结束，关系到qos的级别 0不会调用，1和2都会调用
     * @param token 彻底结束的那条消息的唯一标识
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            System.out.println("消费者已经消费完毕，消息id：" + token.getMessage().getId());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
